package com.sweet;

import com.sweet.entity.ClientAckMessage;
import com.sweet.entity.MessageBody;
import com.sweet.entity.ServerAckMessage;
import com.sweet.enums.MessageType;
import com.sweet.store.MessageStore;
import com.sweet.store.StoreVerticle;
import com.sweet.store.TokenStore;
import com.sweet.store.UserStore;
import io.netty.util.internal.StringUtil;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.json.JsonObject;
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetSocket;
import io.vertx.core.parsetools.RecordParser;
import io.vertx.core.shareddata.LocalMap;

import java.util.HashMap;
import java.util.Map;


/**
 * Created by sweet on 2019/1/14.
 * ---------------------------
 */
public class MainVerticle extends AbstractVerticle {

  public static void main(String[] args) {
    Vertx vertx = Vertx.vertx(new VertxOptions().setBlockedThreadCheckInterval(1000L * 10));
    vertx.deployVerticle(new MainVerticle());
    vertx.deployVerticle(new IMHttpServerVerticle());
  }

  private final String delimit = "\n";

  private final long waitTime = 1000L * 3;

  public static final String PUSH_MESSAGE_ADDRESS = "pushMessage";

  private final long heartbeatTime = 1000L * 60 * 5;
  private final long delay = 1000L * 60 * 2;


  private final Map<String, SocketMapValue> socketMap = new HashMap<>();

  @Override
  public void start() throws Exception {
    super.start();
    // 心跳检查
    heartbeatHandler();
    // 推送由 http server 过来的请求
    httpServerPushMessage();
    NetServer netServer = vertx.createNetServer();
    netServer.connectHandler(netSocket -> {
      System.out.println("socket 上线：" + netSocket.writeHandlerID());
      final RecordParser recordParser = RecordParser.newDelimited(delimit,
              m -> this.messageHandler(m, netSocket));
      netSocket.handler(recordParser);

      netSocket.exceptionHandler(ex -> {
        ex.printStackTrace();
        System.out.println("socket ex: " + ex.getMessage());
      });
      netSocket.closeHandler(close -> {
        System.out.println("socket close: " + netSocket.writeHandlerID());
      });
      netSocket.endHandler(end -> {
        System.out.println("socket end: " + netSocket.writeHandlerID());
      });
    });
    netServer.exceptionHandler(ex -> {
      ex.printStackTrace();
      System.out.println("netServer ex: " + ex.getMessage());
    });
    netServer.listen(9090, server -> {
      if (server.failed()) server.cause().printStackTrace();
      System.out.println("server ok 9090");
    });
  }

  private void write(String msg, String mapKey, NetSocket netSocket) {
    try {
      netSocket.write(msg + delimit);
    } catch (IllegalStateException e) {
      e.printStackTrace();
      System.out.println("socket is close");
      socketMap.remove(mapKey);
    }
  }

  private void httpServerPushMessage() {
    vertx.eventBus().<JsonObject>consumer(PUSH_MESSAGE_ADDRESS, m -> {
      System.out.println(m.body().encodePrettily());
      MessageBody messageBody = MessageBody.fromJson(m.body());
      String to = messageBody.getTo();
      if (socketMap.containsKey(to)) {
        write(m.body().encode(), to, socketMap.get(to).getNetSocket());
      } else {
        UserStore.instance().saveMessage(messageBody.getTo(), m.body().encode(), vertx);
      }
    });
  }

  private void heartbeatHandler() {
    vertx.setPeriodic(delay, (t) -> {
      long nowTime = System.currentTimeMillis();
      Map<String, SocketMapValue> copyMap = new HashMap<>(socketMap);
      copyMap.forEach((userId, socket) -> {
        if (nowTime - socket.getTime() > heartbeatTime) {
          System.out.println("userId: " + userId + " 掉线");
          SocketMapValue remove = socketMap.remove(userId);
          if (remove != null) remove.getNetSocket().close();
        }
      });
    });
  }

  private void messageHandler(Buffer buffer, NetSocket netSocket) {
    try {
      final JsonObject messageJson = buffer.toJsonObject();
      final String token = messageJson.getString("token");
      if (!StringUtil.isNullOrEmpty(token)) {
        TokenStore.instance().findKey(token, vertx)
                .setHandler(obj -> {
                  if (obj.succeeded()) {
                    String userId = obj.result().getUser().getId();
                    socketMap.put(userId, new SocketMapValue(netSocket, System.currentTimeMillis()));
                    UserStore.instance().getOffLineMessage(userId, vertx)
                            .setHandler(r -> {
                              if (r.succeeded()) {
                                r.result().forEach(s -> write(s, userId, netSocket));
                              } else {
                                r.cause().printStackTrace();
                                System.out.println("获取离线消息出错，" + r.cause().getMessage());
                              }
                            });
                    System.out.println("合法消息： " + messageJson.encodePrettily());
                    messageHandler1(messageJson, netSocket);
                  } else {
                    netSocket.close();
                  }
                });
      } else {
        netSocket.close();
      }
    } catch (Exception e) {
      e.printStackTrace();
      System.out.println("解析json出错");
      netSocket.close();
    }


  }

  private void messageHandler1(JsonObject messageJson, NetSocket netSocket) {
    final String type = messageJson.getString("type");
    final MessageType messageType = MessageType.TEXT.name().equals(type) ?
            MessageType.TEXT : (MessageType.ACK.name().equals(type) ? MessageType.ACK : MessageType.PING);
    switch (messageType) {
      case TEXT:
        final MessageBody messageBody = MessageBody.fromJson(messageJson);
        String fromUserId = messageBody.getFrom();
        socketMap.put(fromUserId, socketMap.get(fromUserId).updateTime(System.currentTimeMillis()));
        String to = messageBody.getTo();
        if (socketMap.containsKey(to)) {
          SocketMapValue value = socketMap.get(to);
          write(messageJson.encode(), to, value.getNetSocket());
        } else {
          UserStore.instance().saveMessage(messageBody.getTo(), messageJson.encode(), vertx);
        }
        return;
      case ACK:
        return;
      case PING:
        TokenStore.instance().findKey(messageJson.getString("token"), vertx)
                .setHandler(f -> {
                  if (f.succeeded()) {
                    String userId = f.result().getUser().getId();
                    socketMap.put(userId, socketMap.get(userId).updateTime(System.currentTimeMillis()));
                  } else {
                    f.cause().printStackTrace();
                    System.out.println("ping error: " + f.cause().getMessage());
                  }
                });

        return;
    }
  }

  private void messageHandler0(JsonObject messageJson, NetSocket netSocket) {
    if (messageJson.containsKey("type")) {
      MessageType type = messageJson.getString("type").equals(MessageType.ACK.name())
              ? MessageType.ACK : MessageType.TEXT;
      switch (type) {
        case TEXT:
          textMessageHandler(messageJson, netSocket);
          return;
        case ACK:
          ackMessageHandler(messageJson, netSocket);
          return;
      }
    }
  }

  // 3、4、
  private void ackMessageHandler(JsonObject messageJson, NetSocket netSocket) {
    final ClientAckMessage clientAckMessage = ClientAckMessage.fromJson(messageJson);
    MessageStore.instance().containsMessage(clientAckMessage.getId(), vertx)
            .setHandler(r -> {
              if (r.succeeded() && r.result()) {

              } else {
                r.cause().printStackTrace();
                System.out.println("处理Client ack message 出错，" + r.cause().getMessage());
              }
            });
  }

  private void textMessageHandler(JsonObject messageJson, NetSocket netSocket) {
    final MessageBody messageBody = MessageBody.fromJson(messageJson);
    MessageStore.instance().save(messageBody.getId(), messageJson.encode(), vertx)
            .setHandler(m -> {
              if (m.succeeded()) {
                // 1. 给 client 发 ack 消息
                netSocket.write(ServerAckMessage.create(messageBody.getId(), MessageType.ACK).toJson());
                // 2. 看对方是否在线
                // 3. 转发消息
              } else {
                m.cause().printStackTrace();
                System.out.println("保存消息失败，" + m.cause().getMessage());
              }
            });
  }

}

class SocketMapValue {
  private final NetSocket netSocket;
  private final long time;

  public SocketMapValue(NetSocket netSocket, long time) {
    this.netSocket = netSocket;
    this.time = time;
  }

  public SocketMapValue updateTime(long time) {
    return new SocketMapValue(netSocket, time);
  }

  public NetSocket getNetSocket() {
    return netSocket;
  }

  public long getTime() {
    return time;
  }
}